package cn.iocoder.yudao.module.bpm.message.kafka;

import cn.iocoder.yudao.module.bpm.message.ProcessTaskMessage;
import cn.iocoder.yudao.module.bpm.message.dto.BpmProcessTaskAssignedMessage;
import cn.iocoder.yudao.module.bpm.message.dto.BpmProcessTaskCancelledMessage;
import cn.iocoder.yudao.module.bpm.message.dto.BpmProcessTaskCompletedMessage;
import cn.iocoder.yudao.module.bpm.message.dto.BpmProcessTaskCreatedMessage;
import lombok.extern.log4j.Log4j2;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.concurrent.ExecutionException;

@Log4j2
public class KafkaProcessTaskMessage implements ProcessTaskMessage {

    private final KafkaTemplate<Object, Object> kafkaTemplate;

    private final String processInstanceTaskCreatedTopic;
    private final String processInstanceTaskAssignedTopic;
    private final String processInstanceTaskCleanTopic;
    private final String processInstanceTaskFinishedTopic;

    public KafkaProcessTaskMessage(
            KafkaTemplate<Object, Object> kafkaTemplate,
            String processInstanceTaskCreatedTopic,
            String processInstanceTaskAssignedTopic,
            String processInstanceTaskCleanTopic,
            String processInstanceTaskFinishedTopic
    ) {
        this.kafkaTemplate = kafkaTemplate;
        this.processInstanceTaskCreatedTopic = processInstanceTaskCreatedTopic;
        this.processInstanceTaskAssignedTopic = processInstanceTaskAssignedTopic;
        this.processInstanceTaskCleanTopic = processInstanceTaskCleanTopic;
        this.processInstanceTaskFinishedTopic = processInstanceTaskFinishedTopic;
    }
    @Override
    public void onProcessInstanceTaskCreatedMsg(BpmProcessTaskCreatedMessage bpmProcessTaskCreatedMessage) {
        try {
            kafkaTemplate.send(processInstanceTaskCreatedTopic, null, bpmProcessTaskCreatedMessage).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("[sendKafkaMessage][发送消息({}) 到 Kafka 失败]", bpmProcessTaskCreatedMessage, e);
        }

    }

    @Override
    public void onProcessInstanceTaskAssignedMsg(BpmProcessTaskAssignedMessage bpmProcessTaskAssignedMessage) {
        try {
            kafkaTemplate.send(processInstanceTaskAssignedTopic, null, bpmProcessTaskAssignedMessage).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("[sendKafkaMessage][发送消息({}) 到 Kafka 失败]", bpmProcessTaskAssignedMessage, e);
        }
    }

    @Override
    public void onProcessInstanceTaskCleanMsg(BpmProcessTaskCancelledMessage bpmProcessTaskCancelledMessage) {
        try {
            kafkaTemplate.send(processInstanceTaskCleanTopic, null, bpmProcessTaskCancelledMessage).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("[sendKafkaMessage][发送消息({}) 到 Kafka 失败]", bpmProcessTaskCancelledMessage, e);
        }
    }

    @Override
    public void onProcessInstanceTaskFinishedMsg(BpmProcessTaskCompletedMessage bpmProcessTaskCompletedMessage) {
        try {
            kafkaTemplate.send(processInstanceTaskFinishedTopic, null, bpmProcessTaskCompletedMessage).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("[sendKafkaMessage][发送消息({}) 到 Kafka 失败]", bpmProcessTaskCompletedMessage, e);
        }
    }
}
